package ds_industry_2025.ds.ds_06.tzgc

import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.DoubleType
import org.apache.spark.sql.{SparkSession, functions}
/*
    2、根据第1小题的结果，对其进行聚合，其中对sku_id进行one-hot转换，将其转换为以下格式矩阵：第一列为用户id，其余列名为商品id，
    按照用户id进行升序排序，展示矩阵第一行前5列数据，将结果截图粘贴至客户端桌面【Release\任务C提交结果.docx】中对应的任务序号下。
字段	类型	中文含义	备注
user_id	double	客户key
sku_id0	double	用户是否购买过商品1	若用户购买过该商品，则值为1，否则为0
sku_id1	double	用户是否购买过商品2	若用户购买过该商品，则值为1，否则为0
sku_id2	double	用户是否购买过商品3	若用户购买过该商品，则值为1，否则为0
.....
结果格式如下：
---------------第一行前5列结果展示为---------------
0.0,1.0,0.0,0.0,0.0
 */
object t2 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("t2")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

    import spark.implicits._

    //  todo 读取到特征工程第一题的结果
    val r1=spark.table("tzgc.t2")


    //  todo 将sku_id字段的值改一下格式
    val source = r1.withColumn(
      "sku_id", functions.concat(lit("sku_id"), col("sku_id"))
    )

    //  todo 提取出来所有的skuid
    val skus:Array[String] = source.select("sku_id")
      .distinct()
      .orderBy(
        split(col("sku_id"), "id")(1).cast("int")
      )
      .map(_(0).toString)
      .collect()

    val result = source.groupBy("user_id")
      .pivot("sku_id", skus)
      .agg(lit(1.0))
      .na.fill(0.0)
      .orderBy("user_id")
      .withColumn("user_id",col("user_Id").cast(DoubleType))

    result.show

    println("-------------第一行前5列结果展示为----------------")
    result.limit(1).select(result.columns.slice(0,5).map(col): _*).collect()
      .foreach(
        r => println(r.toSeq.mkString(","))
      )

    //  todo 把结果保存提供给下面的推荐系统
    result.write.format("hive").mode("overwrite")
      .saveAsTable("tzgc.skumat")



    spark.close()

  }

}
